package cn.bawei.shop.realtime.etl.process
import java.io.File

import cn.bawei.canal.util.ip.IPSeeker
import cn.bawei.shop.realtime.etl.`trait`.MQBaseETL
import cn.bawei.shop.realtime.etl.bean.{ClickLogBean, ClickLogWideBean}
import cn.bawei.shop.realtime.etl.utils.DateUtil.datetime2date
import cn.bawei.shop.realtime.etl.utils.{DateUtil, GlobalConfigUtil}
import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializerFeature
import nl.basjes.parse.httpdlog.HttpdLoglineParser
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}

/**
 * 点击流日志的实时ETL操作
 * 需要将点击流日志对象转换成拓宽后的点击流对象，增加省份、城市、时间字段
 * @param env
 */

class ClickLogETL(env:StreamExecutionEnvironment) extends MQBaseETL(env) {
  /**
   * 处理数据的接口
   * 根据业务抽取出来process方法，因为所有的ETL都有操作方法
   */
  override def process(): Unit = {
    /**
     * 实现步骤：
     * 1：获取点击流日志的数据源
     * 2：将nginx的点击流日志字符串转换成点击流对象
     * 3：对点击流对象进行实时拉宽操作，返回拉宽后的点击流实体对象
     * 4：将拉宽后点点击流实体类转换成json字符串
     * 5：将json字符串写入到kafka集群，供Druid进行实时的摄取操作
     */
    // 1. 整合kafka
    //1：获取点击流日志的数据源
    val clickLogDS: DataStream[String] = getKafkaDataSream(GlobalConfigUtil.`input.topic.click_log`)

    clickLogDS.printToErr()
    // 2. 实时ETL
    //2：将nginx的点击流日志字符串转换成点击流对象
    val clickLogWideDataStream: DataStream[ClickLogWideBean] = etl(clickLogDS)
    clickLogWideDataStream.print("点击流日志>>>")

    //3：将clickLogWideBeanDS转换成json字符串返回，因为kafka中需要传入字符串类型的数据
    //4：将拉宽后点点击流实体类转换成json字符串
    val clickLogWideJsonDataStream: DataStream[String] = clickLogWideDataStream.map(clickLogWideEntityEntity => {
      //将拉宽后的点击流对象样例类转换成json字符串
      JSON.toJSONString(clickLogWideEntityEntity, SerializerFeature.DisableCircularReferenceDetect)
    })
    //打印测试
    clickLogWideJsonDataStream.printToErr("拉宽后的点击流日志>>>")

    //5：将关联维度表后的数据写入到kafka中，供Druid进行指标分析
    //5：将json字符串写入到kafka集群，供Druid进行实时的摄取操作
    clickLogWideJsonDataStream.addSink(kafkaProducer(GlobalConfigUtil.`output.topic.clicklog`))
  }

  /**
   * 数据实时ETL
   * @param clicklogDS
   * @return
   */
  /**
   * 将点击流日志字符串转换成拉宽后的点击流对象
   * @param clickLogDataStream
   */
  private def etl(clicklogDS:DataStream[String]) = {
    /**
     * 实现步骤：
     * 1：将点击流日志字符串转换成点击流对象，使用logparsing解析
     * 2：根据ip地址，获取到ip地址对应的省份、城市、访问时间等信息，需要一个ip地址库，传递ip地址返回对应的省份和城市信息
     * 3：将获取到的省份和城市作为拉宽后对象的参数传递进去，将拉宽后的点击流对象返回
     */
    // 1. 基于JSON解析字段
    //1：将点击流日志字符串转换成点击流对象，使用logparsing解析
    val clickLogBeanDS: DataStream[ClickLogBean] = clicklogDS.map(new RichMapFunction[String, ClickLogBean] {
      //定义数据格式化的解析器对象
      var parser: HttpdLoglineParser[ClickLogBean] = null
      //这个方法只被初始化一次，一般用于初v始化外部资源
      override def open(parameters: Configuration): Unit = {
        //实例化解析器
        parser = ClickLogBean.createClickLogParser()
      }
      //对数据一条条的进行转换
      override def map(clickLog: String): ClickLogBean = {
        //将点击流字符串转换成点击流对象返回
        ClickLogBean(parser, clickLog)
      }
    })

    // 2. 使用JavaURL解析URL地址，获取host名称
    clickLogBeanDS.map(new RichMapFunction[ClickLogBean, ClickLogWideBean] {
      var ipSeeker:IPSeeker = _

      override def open(parameters: Configuration): Unit = {
        val dataFile: File = getRuntimeContext.getDistributedCache.getFile("qqwry.dat")
        ipSeeker = new IPSeeker(dataFile)
      }

      override def map(in: ClickLogBean): ClickLogWideBean = {
        // 拷贝原有属性
        val wideBean = ClickLogWideBean(in)
        // wideBean.timestamp = DateUtil.datetime2timestamp(in.getRequestTime)
        val country = ipSeeker.getCountry(wideBean.ip)
        //山东省潍坊市
        var areaArray = country.split("省");
        if(areaArray.length > 1) {
          wideBean.province = areaArray(0) + "省";
          wideBean.city = areaArray(1)
        }
        else {
          areaArray = country.split("市");
          if(areaArray.length > 1) {
            wideBean.province = areaArray(0) + "市";
            wideBean.city = areaArray(1)
          }
          else {
            wideBean.province = areaArray(0);
            wideBean.city = ""
          }
        }
        wideBean.requestDateTime=DateUtil.date2DateStr(datetime2date("05/Sep/2010:11:27:50 +0200"), "yyyy-MM-dd HH:mm:ss")
        wideBean
      }
    })
  }
}
